package com.beiding.jsonanalyzer.flow;

import com.beiding.jsonanalyzer.flow.branchevents.StopNow;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 数据缓冲流
 * 被观察者
 * 内部
 *
 * @author 丁常磊
 * @date 2022/5/12 9:25
 */

class DataFlow {


    /**
     * 当前写入的队列
     * 主线程中操作
     */
    private BlockingQueue<Object> currentActionQueue;


    /**
     * 当前消费中的队列
     * 任务线程中操作
     */
    private BlockingQueue<Object> currentConstActionQueue;

    /**
     * 事件消费者
     * 任务线程中操作
     */
    private Map<Class, EventListener> eventListenerMap = new HashMap<>();

    private BranchNode branchNode;


    /**
     * 历史任务队列
     * 任务线程中操作
     */
    private List<BlockingQueue<Object>> historyActionQueues = new ArrayList<>();

    DataFlow(BranchNode branchNode) {
        this.branchNode = branchNode;
        currentActionQueue = new LinkedBlockingQueue<>();
        currentConstActionQueue = currentActionQueue;
        //注册内部事件处理逻辑
        onInnerEvent();
    }


    /**
     * 订阅给定类型的事件
     *
     * @param type
     * @param eventListener
     */
    synchronized void on(Class type, EventListener eventListener) {

        //放入任务槽任务
        branchNode.getBranch().getTaskSlot().commit(() -> {
            eventListenerMap.put(type, eventListener);
            consume();
        });


    }


    /**
     * 移除消费者
     *
     * @param type
     */
    synchronized void removeListener(Class type) {

        //提交一个槽任务,在当前消费任务执行完成后才会更新观察者
        branchNode.getBranch().getTaskSlot().commit(() -> {
            eventListenerMap.remove(type);
        });

    }

    /**
     * 自己提交,自己消费
     *
     * @param e
     */
    synchronized void emit(Object e) {

        //添加事件
        currentActionQueue.add(e);

        //放到任务槽中执行
        branchNode.getBranch().getTaskSlot().commit(this::consume);

    }

    private void consume() {


        //消费时,先消费完历史队列. 再消费当前队列

        if (historyActionQueues.size() > 0) {
            Iterator<BlockingQueue<Object>> iterator = historyActionQueues.iterator();
            while (iterator.hasNext()) {
                BlockingQueue<Object> next = iterator.next();
                consume(next);
                if (next.size() == 0) {
                    //消费干净,历史队列移除
                    iterator.remove();
                } else {
                    //没有消费干净,不再消费
                    break;
                }
            }
        }

        //再消费当前队列
        consume(currentConstActionQueue);

    }

    private Runnable runConsume = () -> {
        branchNode.getBranch().getTaskSlot().commit(this::consume);
    };

    private void consume(BlockingQueue<Object> actionQueue) {

        //如果没有任务
        if (actionQueue.size() == 0) {
            return;
        }

        //开始执行任务
        Iterator<Object> iterator = actionQueue.iterator();

        while (iterator.hasNext()) {
            Object next = iterator.next();

            EventListener eventListener = eventListenerMap.get(next.getClass());

            //消费者存在则消费,不存在则结束消费
            if (eventListener != null) {
                try {
                    eventListener.on(next);
                } catch (Exception ex) {
                    //发出全局终止信号
                    branchNode.getBranch().emit(StopNow.error(ex));
                } finally {
                    iterator.remove();
                }
            } else {
                break;
            }
        }
    }

    /**
     * 异步处理
     *
     * @param maxSize
     */
    synchronized void maxEventCount(int maxSize) {

        BlockingQueue<Object> ncq;

        if (maxSize <= 0) {
            //设置为无界队列
            ncq = new LinkedBlockingQueue<>();
        } else {
            //设置为有界队列
            ncq = new ArrayBlockingQueue<>(maxSize);
        }

        //获取当前任务队列
        BlockingQueue<Object> oq = currentActionQueue;

        //直接切换
        currentActionQueue = ncq;

        //将当前队列添加到历史队列. 更新当前消费中的队列
        branchNode.getBranch().getTaskSlot().commit(() -> {
            historyActionQueues.add(oq);
            currentConstActionQueue = ncq;
        });

    }

    /**
     * 默认的刷新事件执行逻辑
     */
    private static final EventListener<FlushEvent> ON_FLUSH_EVENT = event -> {
        //直接执行; 除非为外部修改了监听逻辑
        event.getCallback().run();
    };


    private void onInnerEvent() {
        //注册内部事件
        on(FlushEvent.class, ON_FLUSH_EVENT);
    }

}
